package com.reactive.demo;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

public class Main {
    public static void main(String[] args) throws InterruptedException {
//        Mono<Object> objectMono= Mono.just("Hello, World!");
//        objectMono.subscribe(System.out::println);
//        System.out.println("Hello, World!");
//        collectMultimap();
        then();
        Thread.sleep(1000);
    }

    public static void flux() {
        Flux.from(new Publisher<String>() {
            @Override
            public void subscribe(Subscriber<? super String> s) {
                for (int i = 0; i < 10; i++) {
                    s.onNext("hello" + i);
                }
                s.onComplete();
            }
        }).subscribe(System.out::println,
                System.err::println,
                () -> System.out.println("complete"));


    }

    public static void collectMultimap() {
        Flux.just(1, 2, 3, 4, 5)
                .collectMultimap(
                        item -> "key:" + item,
                        item -> {
                            List<String> values = new ArrayList<>();
                            for (int i = 0; i < item; i++) {
                                values.add("value:" + i);
                            }
                            return values;
                        },
                        () -> { // 扩充
                            Map map = new HashMap();
                            map.put("other", "other");
                            return map;
                        })
                .subscribe(System.out::println);


    }

    public static void then() {
        Flux<String> just = Flux.just("Apple", "Orange", "Grape", "Banana", "Strawberry");
// 返回一个Mono ，在此Flux完成时完成。这将主动忽略序列，只重放完成或错误信号。
        just.doOnNext(t -> System.out.println("发布者处理数据：" + t))
                .then(Mono.defer(() -> {
                    return Mono.just("我完成了");
                }))
                .subscribe(t -> System.out.println("订阅者处理数据：" + t));
/**
 * 执行结果：
 * 发布者处理数据：Apple
 * 发布者处理数据：Orange
 * 发布者处理数据：Grape
 * 发布者处理数据：Banana
 * 发布者处理数据：Strawberry
 * 订阅者处理数据：我完成了
 */
    }
}
